package com.ccf.business.etl.websocket;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.ccf.business.etl.model.FlowLog;
import com.ccf.business.etl.service.IFlowLogService;
import com.service.websocket.handler.AbstractMessageHandler;
import com.service.websocket.to.MessageTO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

/**
 * 描述：
 * 流程测试流程处理器
 *
 * ws://127.0.0.1:8221/etl/wjWs?busId=9a50379077514278a748d6cfb6b618be
 *
 * {"topic":"testLog"}
 *
 * @author xianggj
 * @Date 2021/12/8 10:22
 **/
@Component
@Slf4j
public class FlowTestLogHandler extends AbstractMessageHandler {

    @Autowired
    private IFlowLogService flowLogService;

    @Override
    public String produce(MessageTO tuple) throws Exception {

        if (StrUtil.isBlank(tuple.getBusId())){
            return MessageTO.FAILD;
        }
        List<String> logInfo = getLogInfo(tuple.getBusId());
        if (CollUtil.isEmpty(logInfo)){
            tuple.setData("流程等待运行中...");
            sendMessageToBus(tuple);
            return MessageTO.SUCCESS;
        }
        tuple.setData(CollUtil.join(logInfo, "\n"));
        sendMessageToBus(tuple);
        return MessageTO.SUCCESS;
    }

    /**
     * 整理消息
     * @param busId
     */
    private List<String> getLogInfo(String busId) {
        QueryWrapper<FlowLog> qw = new QueryWrapper<>();
        qw.eq("flow_id", busId);
        qw.orderByAsc("log_id");
        List<FlowLog> list = flowLogService.list(qw);
        List<String> res = new ArrayList<>();
        if (CollUtil.isNotEmpty(list)){
            for (FlowLog f : list) {
                res.add(f.toString());
            }
        }
        return res;
    }

    @Override
    public String getTopic() {
        return WebsocketTopic.TEST_LOG.getValue();
    }

    @Override
    public Object getParams(Object map) throws Exception {
        return null;
    }

    @Override
    public void afterConnectionClosed(String winId) {
            log.info(getTopic() + "窗口释放成功:winid:" + winId);
    }

    /**
     * 主动发送消息
     *  这里发送太快 会导致EOF异常
     * @param flowId
     * @throws Exception
     */
    public void sendMessage(String flowId) throws Exception {
        MessageTO messageTO = new MessageTO();
        messageTO.setBusId(flowId);
        messageTO.setTime(System.currentTimeMillis());
        messageTO.setTopic(WebsocketTopic.TEST_LOG.getValue());
        List<String> logInfo = getLogInfo(flowId);
        if (CollUtil.isEmpty(logInfo)){
            log.warn("暂无日志信息");
            messageTO.setData("流程等待运行中...");
            sendMessageToBus(messageTO);
            return;
        } else{
            messageTO.setData(CollUtil.join(logInfo, "\n"));
            sendMessageToBus(messageTO);
        }
    }
}
